package com.weimin.mqtt.server.management;

import com.weimin.mqtt.server.listener.IMqttSessionListener;
import com.weimin.mqtt.server.model.MqttSession;
import com.weimin.mqtt.server.model.MqttSessionPublishMessage;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class MqttSessionManager {

    // 连接ID 为key
    private final Map<String, MqttSession> connectMap = new ConcurrentHashMap<>();
    // 客户端ID 为key
    private final Map<String, MqttSession> clientsMap = new ConcurrentHashMap<>();

    private final IMqttSessionListener mqttSessionListener;

    private final ThreadPoolExecutor messageListenerPool = new ThreadPoolExecutor(
            0, // 最小线程数
            4, // 最大线程数
            60, // 线程存活时间（60秒）
            TimeUnit.SECONDS, // 时间单位
            new LinkedBlockingQueue<>() // 任务队列，这里设置为无线队列
    );

    public MqttSessionManager() {
        mqttSessionListener = null;
    }

    public MqttSessionManager(IMqttSessionListener mqttSessionListener) {
        this.mqttSessionListener = mqttSessionListener;
    }

    public synchronized void offline(String connectId) {
        MqttSession session = connectMap.get(connectId);
        if (session != null) {
            clientsMap.remove(session.getClientId());
            connectMap.remove(connectId);
            log.info("MQTT 用户离线 客户端ID: {} 用户名: {}", session.getClientId(), session.getUsername());
            if (mqttSessionListener != null) {
                messageListenerPool.submit(() -> mqttSessionListener.offline(session.getClientId(), session));
            }
        }
    }

    /**
     * 接收到客户端返回的数据说明设备在线，刷新一下session
     */
    public synchronized void online(String ip, String connectId, Function<MqttSessionPublishMessage, Boolean> publisher, String clientId, String username, byte[] password) {
        MqttSession session = clientsMap.get(clientId);
        if (session == null) {
            log.info("MQTT 用户上线 IP: {} 客户端ID: {} 用户名: {}", ip, clientId, username);
            session = new MqttSession(username, password, ip, clientId, connectId, publisher);
            clientsMap.put(clientId, session);
            connectMap.put(connectId, session);
        } else {
            log.info("MQTT 用户在线,IP: {} 客户端ID: {} 用户名: {} ", ip, clientId, username);
            if (!connectMap.containsKey(connectId)) {
                String oldConnectId = session.getConnectId();
                if (oldConnectId != null && oldConnectId.length() > 0 && !oldConnectId.equals(connectId)) {
                    connectMap.remove(oldConnectId);
                }
                connectMap.put(clientId, session);
            }
            session.setIp(ip);
            session.setPublisher(publisher);
            session.setRefreshTime(System.currentTimeMillis());
            session.setConnectId(connectId);
            session.setClientId(clientId);
            session.setUsername(username);
            session.setPassword(password);
        }
        if (this.mqttSessionListener != null) {
            MqttSession finalSession = session;
            messageListenerPool.submit(() -> mqttSessionListener.online(clientId, finalSession));
        }
    }

    public void subTopics(String connectId, Set<String> topics) {
        MqttSession session = connectMap.get(connectId);
        if (session == null) {
            log.error("错误的订阅连接,用户为空,连接ID: {} 订阅主题内容: {}", connectId, topics);
            return;
        }
        Set<String> subTopics = session.getSubTopics();
        if (subTopics == null) {
            subTopics = topics;
            session.setSubTopics(topics);
        } else {
            subTopics.addAll(topics);
        }
        log.info("MQTT 用户订阅  客户端ID: {} 新主题: {},存在主题: {}", session.getClientId(), topics, subTopics);
        if (this.mqttSessionListener != null) {
            Set<String> finalSubTopics = subTopics;
            messageListenerPool.submit(() -> mqttSessionListener.onSubTopics(session.getClientId(), topics, finalSubTopics));
        }
    }

    public void unSubTopics(String connectId, List<String> topics) {
        MqttSession session = connectMap.get(connectId);
        if (session == null) {
            log.error("错误的取消订阅连接,用户为空,连接ID: {} 订阅主题内容: {}", connectId, topics);
            return;
        }
        if (session.getSubTopics() != null) {
            topics.forEach(session.getSubTopics()::remove);
            log.info("MQTT 用户取消订阅  客户端ID: {} 取消主题: {},存在主题: {}", session.getClientId(), topics, session.getSubTopics());
            if (this.mqttSessionListener != null) {
                messageListenerPool.submit(() -> mqttSessionListener.onUnSubTopics(session.getClientId(), topics, session.getSubTopics()));
            }
        }
    }

    public void onMessage(MqttSessionPublishMessage publishMessage) {
        messageListenerPool.submit(() -> {
            forwardTopicMessage(publishMessage.getTopic(), publishMessage.getContent());
        });
        if (mqttSessionListener != null) {
            messageListenerPool.submit(() -> mqttSessionListener.onMessage(publishMessage));
        }
    }

    public void forwardTopicMessage(String topic, byte[] content) {
        if (connectMap.size() == 0) {
            return;
        }
        List<MqttSession> sendSessionList = connectMap.values().stream().filter((session) -> {
            for (String subTopic : session.getSubTopics()) {
                if (isMatch(topic, subTopic)) {
                    return true;
                }
            }
            return false;
        }).collect(Collectors.toList());
        if (sendSessionList.size() > 0) {
            sendSessionList.forEach((sendSession) -> {
                try {
                    sendSession.getPublisher().apply(new MqttSessionPublishMessage(sendSession.getConnectId(), topic, content));
                } catch (Exception e) {
                    log.error("MQTT 消息发送失败,主题: {},客户端ID: {} 连接ID: {}", topic, sendSession.getClientId(), sendSession.getConnectId());
                }
            });
        }
    }


    public static boolean isMatch(String topic, String pattern) {
        if ((topic == null) || (pattern == null)) {
            return false;
        }
        if (topic.equals(pattern)) {
            return true;
        }
        if ("#".equals(pattern)) {
            return true;
        }
        String[] splitTopic = topic.split("/");
        String[] splitPattern = pattern.split("/");
        boolean match = true;
        // 如果包含 # 则只需要判断 # 前面的
        for (int i = 0; i < splitPattern.length; i++) {
            if (!"#".equals(splitPattern[i])) {
                // 不是# 号 正常判断
                if (i >= splitTopic.length) {
                    // 此时长度不相等 不匹配
                    match = false;
                    break;
                }
                if (!splitTopic[i].equals(splitPattern[i]) && !"+".equals(splitPattern[i])) {
                    // 不相等 且不等于 +
                    match = false;
                    break;
                }
            } else {
                // 是# 号  肯定匹配的
                break;
            }
        }
        return match;
    }

}
